package com.zyx.flinkdemo.stream.statebackend;


import org.apache.flink.configuration.Configuration;
import org.apache.flink.contrib.streaming.state.EmbeddedRocksDBStateBackend;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;

/**
 * @author zyx
 * @since 2021/9/13 14:41
 * desc: 生产环境下rocksdb配置
 * reference:   https://nightlies.apache.org/flink/flink-docs-release-1.13/docs/deployment/config/
 *              https://cloud.tencent.com/developer/article/1592441
 *              https://www.jianshu.com/p/c2ca92693d24
 */
public class ProdRocksdbStateBackendDemo {
    public static void main(String[] args) throws Exception {
        // 配置rocksdb相关参数
        Configuration conf = new Configuration();
        // MemTable 系列参数
        // Write Buffer Size: 控制 MemTable 的阈值,Write Buffer 越大,写放大效应越小,写性能也会改善,默认大小是 64 MB, 可以根据实际情况适当增大
        conf.setString("state.backend.rocksdb.writebuffer.size", "128MB");
        // Write Buffer Count: 控制内存中允许保留的 MemTable 最大个数, 默认为 2, 建议设置到 5 左右。
        conf.setInteger("state.backend.rocksdb.writebuffer.count", 5);

        // Block Cache 系列参数
        // Block Size: 增加该配置导致写入性能增强, 读取性能下降, 需要搭配 Block Cache Size 调整
        // 建议生产环境调整到 16 ~ 32 KB，内存充足可以设为 128 KB
        conf.setString("state.backend.rocksdb.block.blocksize", "128KB");
        // Block Cache Size：增加 Block 该配置可以明显增加读性能, 默认大小为 8 MB,建议设置到 64 ~ 256 MB
        conf.setString("state.backend.rocksdb.block.cache-size", "128MB");

        // Generic参数
        // Max Open Files: 决定了 RocksDB 可以打开的最大文件句柄数, 如果进程的 ulimit 没有限制, 建议改为 -1, 默认值也为-1 (Flink1.13)
        conf.setInteger("state.backend.rocksdb.files.open", -1);

        // Index 和 Bloom Filter 系列参数
        // Cache Index And Filter Blocks: 表示是否在内存里缓存索引和过滤器 Block, 建议在 Key 具有局部热点时打开
        // This option only has an effect when 'state.backend.rocksdb.memory.managed' or 'state.backend.rocksdb.memory.fixed-per-slot' are configured.
        conf.setBoolean("state.backend.rocksdb.memory.partitioned-index-filters", true);
        // Optimize Filter For Hits: 表示是否会给 L0 生成 Bloom Filter, 建议在 Key 具有局部热点时打开
        conf.setBoolean("taskmanager.runtime.hashjoin-bloom-filters", true);

        // Flush 和 Compaction 相关参数
        // Max Bytes For Level Base：表示 L1 层大小阈值
        //      该参数太小, 每层能存放的 SSTable 较少, 导致层级很多, 造成查找困难
        //      该参数太大, 每层 SSTable 较多, 导致执行 Compaction 等操作的耗时较长, 此时容易出现 Write Stall（写停止）现象, 造成写入中断
        //      默认值为256MB, 建议设为 target_file_size 的倍数
        conf.setString("state.backend.rocksdb.compaction.level.max-size-level-base", "");

        // Target File Size: 表示上一级的 SST 文件达到多大时触发 Compaction 操作, 默认值是 64MB（每增加一级, 阈值会自动乘以 target_file_size_multiplier）
        // 为了减少 Compaction 的频率, 生产环境可调整为 128MB
        conf.setString("state.backend.rocksdb.compaction.level.target-file-size-base", "128MB");

        // Thread Num: 表示后台进行 Compaction 和 Flush 操作的线程数, 默认为 1, 生产环境建议调大为 4
        conf.setInteger("state.backend.rocksdb.thread.num", 4);

        // 获取执行环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment(conf);
        env.setParallelism(1);
        // 允许检查点
        env.enableCheckpointing(10 * 1000L);
        // 设置RocksDBStateBackend为状态后端, true表示使用增量快照
        EmbeddedRocksDBStateBackend rocksdbBackend = new EmbeddedRocksDBStateBackend(true);
        env.setStateBackend(rocksdbBackend);
        env.getCheckpointConfig().setCheckpointStorage("hdfs://linux203:8020/test/checkpoint_test");
        // 读取数据源数据并打印
        env.socketTextStream("linux201", 7777).print();
        // 执行环境
        env.execute();
    }
}
